package net.xuele.learn.flink.book;

import net.xuele.learn.flink.book.utils.SensorReading;
import net.xuele.learn.flink.book.utils.SensorSource;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author patrick
 * @Date 2023/7/7 11:23
 * @Description 副输出
 */
public class Output {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<SensorReading> readings = env.addSource(new SensorSource());
        SingleOutputStreamOperator<SensorReading> process = readings
                .process(new FreezingMonitor());


        process.getSideOutput(new OutputTag<String>("freezing-alarm"){})
                        .print();

        readings.print();

        env.execute();
    }

    private static class FreezingMonitor extends ProcessFunction<SensorReading, SensorReading> {

        private static final long serialVersionUID = -5318691447123611048L;

        OutputTag<String> freezingAlarmOutput = new OutputTag<String>("freezing-alarm"){
            private static final long serialVersionUID = -5271229013429607069L;
        };

        @Override
        public void processElement(SensorReading value, ProcessFunction<SensorReading, SensorReading>.Context ctx, Collector<SensorReading> out) throws Exception {
            if (value.temperature < 32.0) {
                ctx.output(freezingAlarmOutput, "freezing-alarm" + value.id);
            }
            out.collect(value);
        }
    }
}
